package com.huacloud.rocketmq;

import org.apache.log4j.Logger;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;


import java.util.List;

/**
 * Created by terrytan on 2017-6-18.
 */
public class NomConsumer {

    private static Logger log = Logger.getLogger(NomConsumer.class);
    private String groupName = null;
    private String nameSrvIp = null;
    private String nameSrvPort = null;
    private String instanceName = null;
    private String topicName = null;
    private String tagname = null;
    DefaultMQPushConsumer consumer = null;

    public NomConsumer(String groupName, String nameSrvIp, String nameSrvPort, String instanceName, String topicName, String tagName) {
        this.groupName = groupName;
        this.nameSrvIp = nameSrvIp;
        this.nameSrvPort = nameSrvPort;
        this.instanceName = instanceName;
        this.topicName = topicName;
        this.tagname = tagName;

        consumer = new DefaultMQPushConsumer(groupName);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setNamesrvAddr(nameSrvIp + ((nameSrvPort.isEmpty()) ? "" : ":" + nameSrvPort));
        consumer.setInstanceName(instanceName);
        consumer.setVipChannelEnabled(false);
        log.info(consumer.getNamesrvAddr());

    }

    public void consumeMSG() {


        try {
            consumer.subscribe(this.topicName, "*");

            consumer.registerMessageListener(new MessageListenerConcurrently() {

                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    //log.info(Thread.currentThread().getName() + " Receive New Messages: " + msgs);

                    if (msgs != null && !msgs.isEmpty()) {
                        for (Message msg : msgs) {
                            log.info(new String(msg.getBody()));
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
